package cloud.xlh.my_system.config;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.stereotype.Component;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.StringJoiner;

/**
 * Created with IntelliJ IDEA.
 *
 * @ project name : xlh_osms
 * @ Author: XuLeHuang
 * @ Date: 2023/2/24 11:53
 * @ Description:
 * 思考：本来想用实现ApplicationContextInitializer接口和spi机制在
 * 准备环境的时候让spring调用initialize方法去主动调用，达到程序启动就运行的目的
 * 但是，忽略了两点：
 * 1.监听binlog是while循环，一旦这么做，主线程就会在这里死循环，后续的bean创建都做不了了，
 * 但是，我新开一个线程，貌似可以解决，问题又出现了
 * 2.监听binlog的时候上次程序启动有些操作没有同步，这时候就会调用redis操作，
 * 而redis对应的模板类还没创建，这时候就会爆发空指针，canal监听就会断开
 */
@Slf4j
@Component
public class CanalClientConfig {

    /*@Resource
    CorpsUploadService corpsUploadService;*/
    @Autowired
    private RedisTemplate redisTemplate;

    @Value("${canal.monitor.mysql.hostName}")
    private String canalMonitorHostName;

    @Value("${canal.monitor.mysql.port}")
    private Integer canalMonitorPort;

    @Value("${canal.monitor.mysql.tableName}")
    private String canalMonitorTableName;

    private static final String SPLIT = ",";

    private final static int BATCH_SIZE = 10000;

    @Autowired
    void asyncInit() {
        new Thread(new Runnable() {
            @Override
            public void run() {
                startMonitorSQL();
            }
        }).start();
    }

    /**
     * 启动服务
     */
    public void startMonitorSQL() {
        CanalConnector connector = CanalConnectors.newSingleConnector(new InetSocketAddress(canalMonitorHostName, canalMonitorPort), "example", "", "");
        try {
            //打开连接
            connector.connect();
            String dataBaseNameStr = doParseDataBaseName();
            log.info("数据库检测连接成功!" + dataBaseNameStr);
            //订阅数据库表,全部表q
            connector.subscribe(canalMonitorTableName);
            //回滚到未进行ack的地方，下次fetch的时候，可以从最后一个没有ack的地方开始拿
            connector.rollback();
            while (true) {
                // 获取指定数量的数据
                Message message = connector.getWithoutAck(BATCH_SIZE);
                long batchId = message.getId();
                int size = message.getEntries().size();
                if (batchId == -1 || size == 0) {
                } else {
                    handleDataChange(message.getEntries());
                }
                // 提交确认
                connector.ack(batchId);
            }
        } catch (Exception e) {
            e.printStackTrace();
            log.error("成功断开监测连接!尝试重连");
        } finally {
            connector.disconnect();
            //防止频繁访问数据库链接: 线程睡眠 10秒
            try {
                log.error("成功断开监测连接!尝试重连");
                Thread.sleep(10 * 1000);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public String doParseDataBaseName() {
        StringJoiner sj = new StringJoiner("," , "[" , "]");
        String[] canalMonitorTableNameArray = canalMonitorTableName.split(SPLIT);
        for (String name :canalMonitorTableNameArray) {
            int lastIndexOf = name.lastIndexOf(".");
            /**
             * 数据库名去重
             **/
            sj.add(name.substring(0 , lastIndexOf));
        }
        return sj.toString();
    }

    /**
     * 打印canal server解析binlog获得的实体类信息
     */
    private void handleDataChange(List<CanalEntry.Entry> entrys) {
        for (CanalEntry.Entry entry : entrys) {
            if (entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONBEGIN || entry.getEntryType() == CanalEntry.EntryType.TRANSACTIONEND) {
                continue;
            }
            //RowChange对象，包含了一行数据变化的所有特征
            CanalEntry.RowChange rowChange;
            try {
                rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());
            } catch (Exception e) {
                throw new RuntimeException("ERROR ## parser of eromanga-event has an error , data:" + entry.toString(), e);
            }
            CanalEntry.EventType eventType = rowChange.getEventType();
            log.info("Canal监测到更新:【database->{}，table->{}】", entry.getHeader().getSchemaName(), entry.getHeader().getTableName());
            switch (eventType) {
                /**
                 * 删除操作
                 */
                case DELETE:
                    System.out.println("canal监控到你删除" + entry.getHeader().getPropsCount() + "记录.....");
                    //corpsUploadService.DeleteOperateToCityInterface(rowChage, entry);
                    break;
                /**
                 * 添加操作
                 */
                case INSERT:
                    System.out.println("canal监控到你写入" + entry.getHeader().getPropsCount() + "记录.....");
                    //corpsUploadService.InsertOperateToCityInterface(rowChage, entry);
                    break;
                /**
                 * 更新操作
                 */
                case UPDATE:
                    System.out.println("canal监控到你更新" + entry.getHeader().getPropsCount() + "记录.....");
                    handleRedis(rowChange);
                    //corpsUploadService.UpdateOperateToCityInterface(rowChage, entry);
                    break;
                default:
                    break;
            }

        }
    }

   /* @Override
    public void run(ApplicationArguments args) throws Exception {
        startMonitorSQL();
    }*/

    public void handleRedis(CanalEntry.RowChange rowChange) {
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new StringRedisSerializer());
        List<CanalEntry.RowData> rowDatasList = rowChange.getRowDatasList();
        for (CanalEntry.RowData rowData : rowDatasList) {
            //一行数据
            List<CanalEntry.Column> oldColumnList = rowData.getBeforeColumnsList();
            for (CanalEntry.Column column : oldColumnList) {
                if (column.getIsKey()) {
//暂时只支持单一主键
                    Long num = redisTemplate.opsForZSet().remove("orderId", column.getValue());
                    if (num != null && num > 0) {//解决多线程消费同一个订单
                        System.out.println(Thread.currentThread().getId() + "线程:" + System.currentTimeMillis() / 1000 + "ms:redis消费了一个任务：消费的订单OrderId为" + column.getValue());
                    }
                    break;
                }
            }
        }
    }
}

